package com.wsjj.gmall.dim.function;

import com.wsjj.gmall.bean.TableProcessDim;
import com.wsjj.gmall.constant.Constant;
import com.wsjj.gmall.util.HbaseUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;

/**
 * 创建hbase中维度表
 */
public class createTableProcess extends RichMapFunction<TableProcessDim,TableProcessDim> {


    Connection hBseConnection = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        hBseConnection = HbaseUtil.getHBseConnection();
    }


    @Override
    public TableProcessDim map(TableProcessDim value) throws Exception {
        System.out.println(111);
        String op = value.getOp();
        String[] split = value.getSinkFamily().split(",");
//                如果配置表删除数据，相当与删除维度表
        if ("d".equals(op)) {
            HbaseUtil.dropHBaseTable(hBseConnection, Constant.HBASE_NAMESPACE, value.getSinkTable());
//                    增加维度表
        } else if ("r".equals(op) || "c".equals(op)) {
            HbaseUtil.createHBaseTable(hBseConnection, Constant.HBASE_NAMESPACE, value.getSinkTable(), split);
        } else {
            HbaseUtil.dropHBaseTable(hBseConnection, Constant.HBASE_NAMESPACE, value.getSinkTable());
            HbaseUtil.createHBaseTable(hBseConnection, Constant.HBASE_NAMESPACE, value.getSinkTable(), split);
        }

        return value;
    }


    @Override
    public void close() throws Exception {
        HbaseUtil.closeHBaseConn(hBseConnection);
    }
}